package com.sloan.community.service;

import com.sloan.community.common.Constant;
import com.sloan.community.common.dto.CommentCountDTO;
import com.sloan.community.common.dto.HotArticleDTO;
import com.sloan.community.common.query.HotArticleQuery;
import com.sloan.community.common.query.SearchQuery;
import com.sloan.community.common.vo.*;
import com.sloan.community.po.ArticleTag;
import com.sloan.community.po.Comment;
import com.sloan.community.po.Tag;
import com.sloan.community.po.User;
import com.sloan.community.repository.*;
import com.sloan.community.util.DateUtils;
import com.sloan.community.util.TextUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.context.annotation.Lazy;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.*;
import java.util.function.Function;

import static com.sloan.community.util.NumberUtils.formatNumber;
import static com.sloan.community.util.TextUtils.HIGH_LIGHT;
import static com.sloan.community.util.TextUtils.TEXT_NUM;

@Slf4j
@Service
@RequiredArgsConstructor(onConstructor_ = {@Lazy})
public class IndexService {

    private final ArticleRepository articleRepository;

    private final CommentRepository commentRepository;

    private final TagRepository tagRepository;

    private final ArticleTagRepository articleTagRepository;

    private final UserRepository userRepository;

    private final DatabaseClient databaseClient;

    // 这里对正文加全文索引
    public Mono<PageVo<IndexArticleVO>> articlePageList(SearchQuery param) {
        String highLight = String.format(HIGH_LIGHT, param.getKeyword());
        Integer offset = (param.getCurrent() - 1) * param.getSize();
        // TODO 以后在替换为ES进行检索，这里先用Mysql Like 匹配
        String querySql = this.createLikeQuerySql(param, offset);
        String countQuerySql = this.createLikeCountQuerySql(param);

        Mono<Long> totalMono = databaseClient.sql(countQuerySql)
                .fetch()
                .one()
                .map(row -> Long.parseLong(String.valueOf(row.get("total"))));

        Flux<IndexArticleVO> articleFlux = databaseClient.sql(querySql)
                .fetch()
                .all()
                .subscribeOn(Schedulers.boundedElastic())
                .publishOn(Schedulers.parallel())
                .map(row -> {
                    IndexArticleVO articleVO = new IndexArticleVO();
                    articleVO.setArticleId(Long.parseLong(row.get("id").toString()));
                    articleVO.setTitle(row.get("title").toString());
                    articleVO.setContent(row.get("content").toString());
                    if (StringUtils.hasText(param.getKeyword())) {
                        articleVO.setTitle(articleVO.getTitle().replace(param.getKeyword(), highLight));
                        articleVO.setContent(TextUtils.highlightText(articleVO.getContent(), param.getKeyword(), TEXT_NUM));
                    } else {
                        articleVO.setContent(articleVO.getContent().substring(0, Math.min(articleVO.getContent().length(), TEXT_NUM)) + "...");
                    }
                    if (Objects.nonNull(row.get("first_picture"))) {
                        articleVO.setFirstPicture(row.get("first_picture").toString());
                    }
                    articleVO.setView(Long.parseLong(row.get("view").toString()));
                    articleVO.setLikeCount(Long.parseLong(row.get("like_count").toString()));
                    if (Objects.nonNull(row.get("created_by"))) {
                        articleVO.setCreatedBy(Long.parseLong(row.get("created_by").toString()));
                    }
                    if (Objects.nonNull(row.get("updated_by"))) {
                        articleVO.setUpdatedBy(Long.parseLong(row.get("updated_by").toString()));
                    }
                    if (Objects.nonNull(row.get("created_time"))) {
                        articleVO.setCreatedTime(DateUtils.transfer(row.get("created_time").toString()));
                    }
                    if (Objects.nonNull(row.get("updated_time"))) {
                        articleVO.setCreatedTime(DateUtils.transfer(row.get("updated_time").toString()));
                    }
                    articleVO.setUserId(Long.parseLong(row.get("user_id").toString()));
                    return articleVO;
                }).cache();

        Flux<Long> userIdFlux = articleFlux.map(IndexArticleVO::getUserId).distinct();
        Mono<Map<Long, User>> userMono = userRepository.findAllById(userIdFlux)
                .collectMap(User::getId, Function.identity());

        Flux<Long> articleIdFlux = articleFlux.map(IndexArticleVO::getArticleId).distinct()
                .switchIfEmpty(Flux.just(0L));
        Flux<ArticleTag> articleTagFlux = articleIdFlux.collectList()
                .flatMapMany(articleTagRepository::findByArticleIdIsIn);

        Mono<Map<Long, Long>> commentCountMapMono = articleIdFlux.collectList()
                .flatMapMany(commentRepository::countByArticleIds)
                .collectMap(CommentCountDTO::getArticleId, CommentCountDTO::getCommentCount);


        Mono<Map<Long, List<Long>>> articleTagMono = articleTagFlux.groupBy(ArticleTag::getArticleId)
                .flatMap(group -> {
                    return group.collectList()
                            .map(item -> new AbstractMap.SimpleEntry<>(group.key(), item));
                })
                .collectMap(AbstractMap.SimpleEntry::getKey,
                        entry -> entry.getValue().stream()
                                .map(ArticleTag::getTagId)
                                .toList());

        Flux<Long> tagIdFlux = articleTagFlux.map(ArticleTag::getTagId);
        Mono<Map<Long, Tag>> tagMono = tagRepository.findAllById(tagIdFlux)
                .collectMap(Tag::getId, Function.identity());

        return Mono.zip(articleFlux.collectList(), userMono, tagMono,
                        articleTagMono, totalMono, commentCountMapMono)
                .map(tuple -> {
                    Map<Long, User> userMap = tuple.getT2();
                    Map<Long, Tag> tagMap = tuple.getT3();
                    Map<Long, List<Long>> articleTagMap = tuple.getT4();
                    for (IndexArticleVO articleVO : tuple.getT1()) {
                        User user = userMap.getOrDefault(articleVO.getUserId(), new User());
                        List<TagVO> list = articleTagMap.get(articleVO.getArticleId())
                                .stream()
                                .map(tagId -> {
                                    Tag tag = tagMap.get(tagId);
                                    TagVO tagVO = new TagVO();
                                    BeanUtils.copyProperties(tag, tagVO);
                                    return tagVO;
                                }).toList();

                        articleVO.setUsername(user.getUsername());
                        articleVO.setNickname(user.getNickname());
                        articleVO.setAvatarUrl(user.getAvatarUrl());
                        articleVO.setAuthFrom(user.getAuthFrom());
                        articleVO.setCommentCount(tuple.getT6().get(articleVO.getArticleId()));
                        articleVO.setFromT(Constant.Oauth2.getByCode(user.getAuthFrom()).getMsg());
                        articleVO.setTags(list);
                    }
                    return new PageVo<IndexArticleVO>(tuple.getT5(), tuple.getT1(),
                            param.getCurrent(), param.getSize(), 6, Boolean.TRUE);
                });
    }

    public Mono<List<IndexTagVO>> tagList() {
        return tagRepository.findAll()
                .subscribeOn(Schedulers.boundedElastic())
                .publishOn(Schedulers.parallel())
                .take(50)
                .map(tag -> {
                    IndexTagVO indexTagVO = new IndexTagVO();
                    BeanUtils.copyProperties(tag, indexTagVO);
                    return indexTagVO;
                }).collectList();
    }

    private String createLikeQuerySql(SearchQuery searchQuery, Integer offset) {
        /**
         * SELECT
         *   ta.*,
         *   tac.extract_text AS content
         * FROM
         *   t_article ta
         *   LEFT JOIN t_article_content tac ON ta.id = tac.article_id
         * WHERE
         *   (ta.title LIKE '%搞笑%'
         *   OR tac.content LIKE '%搞笑%')
         * ORDER BY ta.created_time DESC
         * LIMIT 0, 10
         */

        StringBuilder sb = new StringBuilder();
        sb.append("SELECT ta.*, tac.extract_text AS content ")
                .append("FROM t_article ta ")
                .append("LEFT JOIN t_article_content tac ON ta.id = tac.article_id ")
                .append("WHERE 1=1 ");

        String keyword = searchQuery.getKeyword();
        if (StringUtils.hasText(keyword)) {
            sb.append("AND (ta.title LIKE '%")
                    .append(keyword)
                    .append("%' ")
                    .append("OR tac.content LIKE '%")
                    .append(keyword)
                    .append("%') ");
        }

        sb.append("ORDER BY ta.created_time DESC ")
                .append("LIMIT ")
                .append(offset)
                .append(", ")
                .append(searchQuery.getSize());

        String sql = sb.toString();
        log.debug("查询首页SQL:{}", sql);
        return sql;
    }


    private String createLikeCountQuerySql(SearchQuery searchQuery) {
        /**
         * SELECT
         *   COUNT(*) AS total
         * FROM
         *   t_article ta
         *   LEFT JOIN t_article_content tac ON ta.id = tac.article_id
         * WHERE
         *   (ta.title LIKE '%搞笑%'
         *   OR tac.content LIKE '%搞笑%')
         */

        StringBuilder sb = new StringBuilder();
        sb.append("SELECT COUNT(*) AS total ")
                .append("FROM t_article ta ")
                .append("LEFT JOIN t_article_content tac ON ta.id = tac.article_id ")
                .append("WHERE 1=1 ");

        String keyword = searchQuery.getKeyword();
        if (StringUtils.hasText(keyword)) {
            sb.append("AND (ta.title LIKE '%")
                    .append(keyword)
                    .append("%' ")
                    .append("OR tac.content LIKE '%")
                    .append(keyword)
                    .append("%') ");
        }

        String sql = sb.toString();
        log.debug("查询count SQL:{}", sql);
        return sql;
    }


    private String createQuerySql(SearchQuery searchQuery, Integer offset) {
        /**
         * SELECT
         *   ta.*,
         *   tac.content
         * FROM
         *   t_article ta
         *   LEFT JOIN t_article_content tac ON ta.id = tac.article_id
         * WHERE
         *   ta.title LIKE '%搞笑%'
         *   OR MATCH(tac.content) AGAINST('搞笑' IN NATURAL LANGUAGE MODE)
         *   ORDER BY ta.created_time DESC
         *   LIMIT 0,
         *   10
         */

        StringBuilder sb = new StringBuilder();
        sb.append("SELECT ta.*, tac.content ")
                .append("FROM t_article ta ")
                .append("LEFT JOIN t_article_content tac ON ta.id = tac.article_id ")
                .append("WHERE 1=1 ");

        String keyword = searchQuery.getKeyword();
        if (StringUtils.hasText(keyword)) {
            sb.append("AND ta.title LIKE '%")
                    .append(keyword)
                    .append("%' ");
        }
        if (StringUtils.hasText(keyword)) {
            sb.append("OR MATCH(tac.content) AGAINST('")
                    .append(keyword)
                    .append("' IN NATURAL LANGUAGE MODE) ");
        }
        sb.append("ORDER BY ta.created_time DESC ")
                .append("LIMIT ")
                .append(offset)
                .append(", ")
                .append(searchQuery.getSize());

        String sql = sb.toString();
        log.debug("查询首页SQL:{}", sql);
        return sql;
    }

    private String createCountQuerySql(SearchQuery searchQuery) {
        /**
         * SELECT
         *   COUNT( * ) AS total
         * FROM
         *   t_article ta
         *   LEFT JOIN t_article_content tac ON ta.id = tac.article_id
         * WHERE
         *   1 = 1
         *   AND ta.title LIKE '%搞笑%'
         *   OR MATCH ( tac.content ) AGAINST ( '搞笑' IN NATURAL LANGUAGE MODE )
         */

        StringBuilder sb = new StringBuilder();
        sb.append("SELECT COUNT(*) AS total ")
                .append("FROM t_article ta ")
                .append("LEFT JOIN t_article_content tac ON ta.id = tac.article_id ")
                .append("WHERE 1=1 ");

        String keyword = searchQuery.getKeyword();
        if (StringUtils.hasText(keyword)) {
            sb.append("AND ta.title LIKE '%")
                    .append(keyword)
                    .append("%' ");
        }
        if (StringUtils.hasText(keyword)) {
            sb.append("OR MATCH(tac.content) AGAINST('")
                    .append(keyword)
                    .append("' IN NATURAL LANGUAGE MODE) ");
        }

        String sql = sb.toString();
        log.debug("查询count SQL:{}", sql);
        return sql;
    }

    /**
     * TODO 最好应该整合flink的“滑动窗口”来解决 排行(TopN)问题
     * hotCount= (5*likeCount + 2*commentCount + 1*view + 3*shareCount ) * MAX(0,1-0.002 * hour(now-createdTime))
     *
     * @return
     */
    public Mono<List<IndexHotVO>> hotList() {
        return articleRepository.hotList(HotArticleQuery.defaultObj())
                .switchIfEmpty(Flux.empty())
                .index()
                .flatMap(tuple -> {
                    HotArticleDTO t2 = tuple.getT2();
                    return userRepository.findById(t2.getUserId())
                            .map(user -> {
                                IndexHotVO indexHotVO = new IndexHotVO();
                                indexHotVO.setIndex(tuple.getT1() + 1);
                                BeanUtils.copyProperties(t2, indexHotVO);
                                indexHotVO.setHotCountText(formatNumber(t2.getHotCount()));
                                indexHotVO.setUsername(user.getUsername());
                                return indexHotVO;
                            });
                })
                .collectList()
                .doOnNext(item -> item.sort(Comparator.comparing(IndexHotVO::getIndex)));
    }

}
